using System.Collections.Concurrent;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace Aurora.RabbitMq.Provider;

/// <summary>
/// 消息消费
/// </summary>
public class RabbitMqMessageConsumer : IRabbitMqMessageConsumer, IDisposable
{
    public ILogger<RabbitMqMessageConsumer> Logger { get; set; }
    protected IConnectionPool ConnectionPool { get; }
    protected AsyncTimer Timer { get; }

    protected ExchangeDeclareConfiguration Exchange { get; private set; } = default!;

    protected QueueDeclareConfiguration Queue { get; private set; } = default!;


    protected string? ConnectionName { get; private set; }


    protected ConcurrentBag<Func<IModel, BasicDeliverEventArgs, Task>> Callbacks { get; }

    protected IModel? Channel { get; private set; }

    protected ConcurrentQueue<QueueBindCommand> QueueBindCommands { get; }


    protected object ChannelSendSyncLock { get; } = new object();


    public RabbitMqMessageConsumer(IConnectionPool connectionPool, AsyncTimer timer)
    {
        Timer = timer;
        ConnectionPool = connectionPool;
        Logger = NullLogger<RabbitMqMessageConsumer>.Instance;
        QueueBindCommands = new ConcurrentQueue<QueueBindCommand>();
        Callbacks = new ConcurrentBag<Func<IModel, BasicDeliverEventArgs, Task>>();
        Timer.Period = 5000; //5 sec.
        Timer.Elapsed = Timer_Elapsed;
        Timer.RunOnStart = true;
    }

    protected virtual async Task Timer_Elapsed(AsyncTimer timer)
    {
        if (Channel == null || Channel.IsOpen == false)
        {
            await TryCreateChannelAsync();
            await TrySendQueueBindCommandsAsync();
        }
    }

    public void Initialize(
        ExchangeDeclareConfiguration exchange,
        QueueDeclareConfiguration queue,
        string? connectionName = null)
    {
        Exchange = exchange == null ? throw new ArgumentNullException(nameof(exchange)) : exchange;
        Queue = queue == null ? throw new ArgumentNullException(nameof(queue)) : queue;
        ConnectionName = connectionName;
        Timer.Start();
    }

    public virtual async Task BindAsync(string routeingKey)
    {
        QueueBindCommands.Enqueue(new QueueBindCommand(QueueBindType.Bind, routeingKey));
        await TrySendQueueBindCommandsAsync();
    }

    public virtual async Task UnbindAsync(string routeingKey)
    {
        QueueBindCommands.Enqueue(new QueueBindCommand(QueueBindType.Unbind, routeingKey));
        await TrySendQueueBindCommandsAsync();
    }

    public virtual void MessageReceived(Func<IModel, BasicDeliverEventArgs, Task> callback)
    {
        Callbacks.Add(callback);
    }

    public virtual void Dispose()
    {
        Timer.Stop();
        DisposeChannel();
    }

    protected virtual void DisposeChannel()
    {
        if (Channel == null)
        {
            return;
        }

        try
        {
            Channel.Dispose();
        }
        catch (Exception ex)
        {
            Logger.LogException(ex, LogLevel.Warning);
            throw;
        }
    }

    /// <summary>
    /// 尝试 queue 绑定/解绑命令
    /// </summary>
    /// <exception cref="ArgumentException"></exception>
    protected virtual  Task TrySendQueueBindCommandsAsync()
    {
        try
        {
            while (!QueueBindCommands.IsEmpty)
            {
                if (Channel == null || Channel.IsClosed)
                {
                    return Task.CompletedTask;
                }

                lock (ChannelSendSyncLock)
                {
                    if (QueueBindCommands.TryPeek(out var command))
                    {
                        switch (command.Type)
                        {
                            case QueueBindType.Bind:
                                Channel.QueueBind(queue: Queue.QueueName, exchange: Exchange.ExchangeName,
                                    routingKey: command.RoutingKey);
                                break;
                            case QueueBindType.Unbind:
                                Channel.QueueUnbind(queue: Queue.QueueName, exchange: Exchange.ExchangeName,
                                    routingKey: command.RoutingKey);
                                break;
                            default:
                                throw new ArgumentException($"Unknown QueueBindType: {command.Type}");
                        }

                        QueueBindCommands.TryDequeue(out command);
                    }
                }
            }
        }
        catch (Exception e)
        {
            Logger.LogException(e, LogLevel.Warning);
            throw;
        }
        return Task.CompletedTask;
    }


    protected virtual async Task TryCreateChannelAsync()
    {
        await DisposeChannelAsync();
        try
        {
            Channel = ConnectionPool.Get(ConnectionName).CreateModel();
            Channel.ExchangeDeclare(exchange: Exchange.ExchangeName, type: Exchange.Type, durable: Exchange.Durable,
                autoDelete: Exchange.AutoDelete, arguments: Exchange.Arguments);

            Channel.QueueDeclare(queue: Queue.QueueName, durable: Queue.Durable, exclusive: Queue.Exclusive,
                autoDelete: Queue.AutoDelete, arguments: Queue.Arguments);

            // 预取 设置 qos
            if (Queue.PrefetchCount.HasValue)
            {
                Channel.BasicQos(prefetchSize: 0, prefetchCount: Queue.PrefetchCount.Value, false);
            }

            var consumer = new AsyncEventingBasicConsumer(Channel);
            consumer.Received += HandleIncomingMessageAsync;

            Channel.BasicConsume(queue: Queue.QueueName, autoAck: false, consumer: consumer);
        }
        catch (Exception e)
        {
            Logger.LogException(e, LogLevel.Warning);
            throw;
        }
    }

    /// <summary>
    /// 释放channel
    /// </summary>
    protected virtual  Task DisposeChannelAsync()
    {
        if (Channel == null)
        {
            return Task.CompletedTask;
        }

        try
        {
            Channel.Dispose();
        }
        catch (Exception e)
        {
            Logger.LogException(e, LogLevel.Warning);
            throw;
        }
        return Task.CompletedTask;
    }


    /// <summary>
    /// 处理 即将达到的消息
    /// </summary>
    /// <param name="sender"></param>
    /// <param name="basicDeliverEventArgs"></param>
    protected virtual async Task HandleIncomingMessageAsync(object sender, BasicDeliverEventArgs basicDeliverEventArgs)
    {
        try
        {
            foreach (var callback in Callbacks)
            {
                await callback(Channel!, basicDeliverEventArgs);
            }

            Channel?.BasicAck(basicDeliverEventArgs.DeliveryTag, multiple: false);
        }
        catch (Exception e)
        {
            try
            {
                Channel?.BasicNack(basicDeliverEventArgs.DeliveryTag, multiple: false, requeue: true);
            }
            catch (Exception exception)
            {
                Logger.LogException(exception, LogLevel.Warning);
            }

            Logger.LogException(e, LogLevel.Warning);
            throw;
        }
    }
}